package com.origin.src.mq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;

import java.nio.charset.StandardCharsets;

public class RocketMq_XAConsumer {

    public static void main(String[] args) throws MQClientException {

        for (int i = 0; i < 3; i++) {
            int j = i;
            new Thread(() -> {
                try {
                    consumer("tx_group" + j);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }

    }

    static ConsumeConcurrentlyStatus status = ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

    public static DefaultMQPushConsumer consumer(String group) throws Exception {
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(group);
        defaultMQPushConsumer.setNamesrvAddr(RocketMQ_XA.nameServer);
        defaultMQPushConsumer.subscribe(RocketMQ_XA.topicName, RocketMQ_XA.tag_XA_Order);
        defaultMQPushConsumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {

            list.forEach((msg) -> {
                String s = ConsoleColorFont.msg(group + " " + Thread.currentThread().getName(), new String(msg.getBody(), StandardCharsets.UTF_8));
                System.out.println(s);
            });
            return status;
        });
        defaultMQPushConsumer.start();
        return defaultMQPushConsumer;
    }
}
